"""
Case Type   : Ustore_Standby_logicreplication
Case Name   : 不同数据库解码
Create At   : 2022/03/29
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建数据库，创建表，创建用户,配置白名单
    5.创建逻辑复制槽
    6.开启逻辑复制槽
    7.运行DDL语句
    8.停止逻辑复制
    9.比对数据
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建数据库成功，创建表成功，创建用户成功，配置白名单成功
    5.创建逻辑复制槽成功
    6.开启逻辑复制槽成功
    7.运行DDL语句，合理报错
    8.停止逻辑复制槽成功
    9.不会解码其它数据库的DML语句
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbylogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0095/')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name_1 = 'db_replication_0095_1'
        self.user_name = 'u_standby_logic_replication_0095'
        self.table_name_db1 = 't_standby_logic_replication_0095_db1'
        self.original_table = 'standby_logic_support_type_tab'
        self.slot_name_1_db1 = 'slot_standby_logic_replication_0095_1_db1'
        self.slot_name_2_db1 = 'slot_standby_logic_replication_0095_2_db1'
        self.log_file_1_db1 = os.path.join(self.dir_replication,
                                           'logic_replication_0095_1_db1.log')
        self.log_file_2_db1 = os.path.join(self.dir_replication,
                                           'logic_replication_0095_2_db1.log')
        self.copy_file_path_db1 = os.path.join(self.dir_replication,
                                               'copy0095_db1.csv')
        self.db_name_2 = 'db_replication_0095_2'
        self.table_name_db2 = 't_standby_logic_replication_0095_db2'
        self.slot_name_1_db2 = 'slot_standby_logic_replication_0095_1_db2'
        self.slot_name_2_db2 = 'slot_standby_logic_replication_0095_2_db2'
        self.log_file_1_db2 = os.path.join(self.dir_replication,
                                           'logic_replication_0095_1db2.log')
        self.log_file_2_db2 = os.path.join(self.dir_replication,
                                           'logic_replication_0095_2_db2.log')
        self.copy_file_path_db2 = os.path.join(self.dir_replication,
                                               'copy0095_db2.csv')
        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        self.logger.info(cp_cmd)
        cp_primy_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_primy_msg)
        self.assertEqual("OK", cp_primy_msg, "执行失败" + text)
        cp_standby_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_standby_msg)
        self.assertEqual("OK", cp_standby_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建数据库，创建表，创建用户,配置白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop database if exists {self.db_name_1};
            create database {self.db_name_1} template template0 
            encoding 'UTF8' lc_ctype 'en_US.UTF-8' lc_collate 'en_US.UTF-8';
            drop database if exists {self.db_name_2};
            create database {self.db_name_2} template template0 
            encoding 'UTF8' lc_ctype 'en_US.UTF-8' lc_collate 'en_US.UTF-8';
            drop user if exists {self.user_name};
            create user {self.user_name} sysadmin 
            password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        msg = create_msg.count(self.constant.DROP_DATABASE_SUCCESS) == 2 and \
              create_msg.count(self.constant.CREATE_DATABASE_SUCCESS) == 2
        self.assertTrue(msg, '执行失败' + text)

        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        self.common.scp_file(self.primary_node, 'table_and_data.sql',
                             self.dir_replication)
        self.sql_file_path = os.path.join(self.dir_replication,
                                          'table_and_data.sql')
        shell_cmd = f"chmod 777 {self.sql_file_path};"
        execute_msg = self.primary_node.sh(shell_cmd).result()
        self.logger.info(execute_msg)
        self.assertNotIn("ERROR", execute_msg, '执行失败' + text)
        db_names = [self.db_name_1, self.db_name_2]
        for db_name in db_names:
            execute_cmd = f"source {macro.DB_ENV_PATH};" \
                f"gsql -d {db_name} " \
                f"-p {self.primary_node.db_port} " \
                f"-f {self.sql_file_path}"
            self.logger.info(execute_cmd)
            execute_msg = self.primary_node.sh(execute_cmd).result()
            self.logger.info(execute_msg)
            self.assertNotIn("ERROR", execute_msg, '执行失败' + text)
        create_cmd = f'''drop table if exists {self.table_name_db1};
                create table {self.table_name_db1} 
                (like {self.original_table});
                alter table {self.table_name_db1} replica identity default;'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname=self.db_name_1)
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS and
                      self.constant.ALTER_TABLE_MSG,
                      create_msg, '执行失败:' + text)
        create_cmd = f'''drop table if exists {self.table_name_db2};
                create table {self.table_name_db2} 
                (like {self.original_table});
                alter table {self.table_name_db2} replica identity default;'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname=self.db_name_2)
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS and
                      self.constant.ALTER_TABLE_MSG,
                      create_msg, '执行失败:' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        db_names = [self.db_name_1, self.db_name_2]
        for db_name in db_names:
            if db_name == self.db_name_1:
                names = [self.slot_name_1_db1, self.slot_name_2_db1]
                for name in names:
                    create_slot = f"pg_recvlogical " \
                        f"-o skip-empty-xacts=true " \
                        f"-d {db_name} -S {name} " \
                        f"-p {self.primary_node.db_port} " \
                        f"-U {self.user_name} -s 2 --create"
                    execute_cmd = f'''source {macro.DB_ENV_PATH};
                            expect <<EOF
                            set timeout 300
                            spawn {create_slot}
                            expect "Password:"
                            send "{macro.COMMON_PASSWD}\\n"
                            expect eof\n''' + '''EOF'''
                    self.logger.info(execute_cmd)
                    execute_msg = self.primary_node.sh(execute_cmd).result()
                    self.logger.info(execute_msg)
            else:
                names = [self.slot_name_1_db2, self.slot_name_2_db2]
                for name in names:
                    create_slot = f"pg_recvlogical " \
                        f"-o skip-empty-xacts=true " \
                        f"-d {db_name} -S {name} " \
                        f"-p {self.primary_node.db_port} " \
                        f"-U {self.user_name} -s 2 --create"
                    execute_cmd = f'''source {macro.DB_ENV_PATH};
                            expect <<EOF
                            set timeout 300
                            spawn {create_slot}
                            expect "Password:"
                            send "{macro.COMMON_PASSWD}\\n"
                            expect eof\n''' + '''EOF'''
                    self.logger.info(execute_cmd)
                    execute_msg = self.primary_node.sh(execute_cmd).result()
                    self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1_db2 and self.slot_name_2_db2 and
                      self.slot_name_1_db2 and self.slot_name_2_db2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical  " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name_1} " \
            f"-S {self.slot_name_1_db1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1_db1} " \
            f"-U {self.user_name} --start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {primy_start_cmd} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1_db1}%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name_1} " \
            f"-S {self.slot_name_2_db1} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2_db1} " \
            f"-U {self.user_name} --start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {stanby_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2_db1}%';"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        primy_start_cmd = f"pg_recvlogical  " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name_2} " \
            f"-S {self.slot_name_1_db2} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1_db2} " \
            f"-U {self.user_name} --start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                                  expect <<EOF 
                                  set timeout 300 
                                  spawn {primy_start_cmd} 
                                  expect "Password:" 
                                  send "{macro.COMMON_PASSWD}\\n" 
                                  expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1_db2}%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d {self.db_name_2} " \
            f"-S {self.slot_name_2_db2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2_db2} " \
            f"-U {self.user_name} --start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                                expect <<EOF 
                                set timeout 300 
                                spawn {stanby_start_cmd} 
                                expect "Password:" 
                                send "{macro.COMMON_PASSWD}\\n" 
                                expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2_db2}%';"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 't', '(1 row)', flag='1')

        text = "-----step7:运行DML语句;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f'''call logical_dml_func('{self.table_name_db1}');
            call insert_to_table(1,3,'{self.table_name_db1}');
            copy {self.table_name_db1} to '{self.copy_file_path_db1}';
            delete from {self.table_name_db1} where col_tinyint is not null;
            copy {self.table_name_db1} from '{self.copy_file_path_db1}';'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg)
        self.assertTrue(sql_msg.count('COPY') == 2, '执行失败:' + text)
        self.assertIn("logical_dml_func" and 'insert_to_table' and
                      self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        sql_cmd = f'''call logical_dml_func('{self.table_name_db2}');
            call insert_to_table(1,3,'{self.table_name_db2}');
            copy {self.table_name_db2} to '{self.copy_file_path_db2}';
            delete from {self.table_name_db2} where col_tinyint is not null;
            copy {self.table_name_db2} from '{self.copy_file_path_db2}';'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg)
        self.assertTrue(sql_msg.count('COPY') == 2, '执行失败:' + text)
        self.assertIn("logical_dml_func" and 'insert_to_table' and
                      self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        time.sleep(5)

        text = "-----step8:停止逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)
        text = "-----验证停止逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1_db1}%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 'f', '(1 row)', flag='1')
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2_db1}%';"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 'f', '(1 row)', flag='1')
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1_db2}%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 'f', '(1 row)', flag='1')
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2_db2}%';"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg, 'active', 'f', '(1 row)', flag='1')

        text = "-----step9:比对数据;expect:不会解码其它数据库的DML语句-----"
        self.logger.info(text)
        shell_cmd_1 = f"sleep 10 && cat {self.log_file_1_db1} " \
            f"|grep {self.table_name_db2} && " \
            f"cat {self.log_file_1_db2} |grep {self.table_name_db1}"
        self.logger.info(shell_cmd_1)
        shell_msg_1 = self.primary_node.sh(shell_cmd_1).result()
        self.logger.info(shell_msg_1)
        self.assertIn("", shell_msg_1, "执行失败" + text)
        shell_cmd_2 = f"sleep 10 && cat {self.log_file_2_db1} " \
            f"|grep {self.table_name_db2} && " \
            f"cat {self.log_file_2_db2} |grep {self.table_name_db1}"
        self.logger.info(shell_cmd_2)
        shell_msg_2 = self.standby_node.sh(shell_cmd_2).result()
        self.logger.info(shell_msg_2)
        self.assertIn("", shell_msg_2, "执行失败" + text)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        db_names = [self.db_name_1, self.db_name_2]
        for db_name in db_names:
            if db_name == self.db_name_1:
                names = [self.slot_name_1_db1, self.slot_name_2_db1]
                for name in names:
                    drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                        f"-d {db_name} -S {name} " \
                        f"-p {self.primary_node.db_port} " \
                        f"-U {self.user_name} -s 2 --drop"
                    execute_cmd = f'''source {macro.DB_ENV_PATH};
                                expect <<EOF
                                set timeout 300
                                spawn {drop_slot}
                                expect "Password:"
                                send "{macro.COMMON_PASSWD}\\n"
                                expect eof\n''' + '''EOF'''
                    self.logger.info(execute_cmd)
                    execute_msg = self.primary_node.sh(execute_cmd).result()
                    self.logger.info(execute_msg)
            else:
                names = [self.slot_name_1_db2, self.slot_name_2_db2]
                for name in names:
                    drop_slot = f"pg_recvlogical -o skip-empty-xacts=true " \
                        f"-d {db_name} -S {name} " \
                        f"-p {self.primary_node.db_port} " \
                        f"-U {self.user_name} -s 2 --drop"
                    execute_cmd = f'''source {macro.DB_ENV_PATH};
                                expect <<EOF
                                set timeout 300
                                spawn {drop_slot}
                                expect "Password:"
                                send "{macro.COMMON_PASSWD}\\n"
                                expect eof\n''' + '''EOF'''
                    self.logger.info(execute_cmd)
                    execute_msg = self.primary_node.sh(execute_cmd).result()
                    self.logger.info(execute_msg)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg_1 = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_1)
        self.logger.info(sql_msg_1)
        sql_msg_2 = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name_2)
        self.logger.info(sql_msg_2)

        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop database {self.db_name_1};
            drop database {self.db_name_2};
            drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]];
            then rm -rf {self.postgresql_path};
            mv {self.postgresql_bak_path} {self.postgresql_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step5.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.primary_node.sh(del_cmd)
        self.standby_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standby_file_msg = self.standby_node.sh(file_cmd).result()
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_3)
        self.assertNotIn(self.slot_name_1_db1 and self.slot_name_2_db1,
                         sql_msg_1, '执行失败:' + text)
        self.assertNotIn(self.slot_name_1_db2 and self.slot_name_2_db2,
                         sql_msg_2, '执行失败:' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_standby_2, '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")